package com.flinksql.test;

import com.flinksql.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author: Lin
 * @create: 2021-07-07 15:24
 * @description:
 **/
public class FlinkSQL_Test6_GroupWindow {
    public static void main(String[] args) throws Exception {

        //TUMBLE滚动  HOP滑动(参数位置注意，time_attr, interval(步长), interval(窗口)) SESSION会话
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("create table sensor(" +
                "id string," +
                "ts bigint," +
                "vc int," +
                "t as to_timestamp(from_unixtime(ts/1000,'yyyy-MM-dd HH:mm:ss'))," +
                "watermark for t as t - interval '5' second" +
                ")with(" +
                "'connector' = 'filesystem'," +
                "'path' = 'input/sensor.txt'," +
                "'format' = 'csv')");

        tableEnv.sqlQuery("SELECT" +
                "  id," +
                "  TUMBLE_START(t, INTERVAL '1' minute) AS wStart," +
                "  TUMBLE_END(t, INTERVAL '1' minute) AS wEnd," +
                "  SUM(vc) sum_vc FROM sensor" +
                "  GROUP BY" +
                "  TUMBLE(t, INTERVAL '1' minute)," +
                "  id")
                .execute()
                .print();

        //env.execute();
    }
}
